
/*
    服务器搭建
*/
#ifndef __M_BROKER_H__
#define __M_BROKER_H__ 
#include "muduo/prot/codec.h"
#include "muduo/prot/dispatcher.h"

#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "muduo/net/TcpConnection.h"

#include "mq_virtualhost.hpp"
#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "../mqcommon/thread_pool.hpp"

namespace xgmq
{
    class Server
    {
    public:
#define DBFILE "mata/dbfile"
#define HOST "host"
        using MessagePtr = std::shared_ptr<google::protobuf::Message>;
        Server(int port, const std::string &basedir)
            : _TcpServer(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server",
                         muduo::net::TcpServer::kReusePort),
              _dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1,
                                    std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1,
                                                               std::placeholders::_2, std::placeholders::_3))),
              _host(std::make_shared<VirtualHost>(HOST,basedir, basedir + DBFILE)),
              _conn_manager(std::make_shared<ConnectionManager>()),
              _consume_manager(std::make_shared<ConsumerManager>()),
              _pool(std::make_shared<ThreadPool>())
        {
            // 服务器运行起来，恢复内存中的队列信息
            // 此时就需要初始化队列的消费者管理结构
            // 先获取所有的消息
            MsgQueueMap qm = _host->allQueues();
            for (auto &q : qm)
            {
                _consume_manager->initQueueConsumer(q.first);
            }
            // 给分发器注册方法,这里需要指明类型
            _dispatcher.registerMessageCallback<openChannelRequest>(std::bind(&Server::OnOpenChannel, this, std::placeholders::_1,
                                                                              std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<closeChannelRequest>(std::bind(&Server::OncloseChannel, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&Server::OndeclareExchange, this, std::placeholders::_1,
                                                                                  std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&Server::OndeleteExchange, this, std::placeholders::_1,
                                                                                 std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareQueueRequest>(std::bind(&Server::OndeclareQueue, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteQueueRequest>(std::bind(&Server::OndeleteQueue, this, std::placeholders::_1,
                                                                              std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueBindRequest>(std::bind(&Server::OnBindQueue, this, std::placeholders::_1,
                                                                            std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueUnBindRequest>(std::bind(&Server::OnUnBindQueue, this, std::placeholders::_1,
                                                                              std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&Server::OnBasicPublish, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&Server::OnBasicAck, this, std::placeholders::_1,
                                                                           std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeRequest>(std::bind(&Server::OnBasicConsume, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&Server::OnBasicCancel, this, std::placeholders::_1,
                                                                              std::placeholders::_2, std::placeholders::_3));
            // 给服务器设置回调函数
            // 收到消息后的回调函数
            _TcpServer.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1,
                                                    std::placeholders::_2, std::placeholders::_3));
            // 连接变化的回调函数
            _TcpServer.setConnectionCallback(std::bind(&Server::OnConnection, this, std::placeholders::_1));
        }
        void start()
        {
            _TcpServer.start();
            _baseloop.loop();
        }

    private:
        // 连接发生变化时的回调函数
        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _conn_manager->newConnection(_host, conn, _codec, _consume_manager, _pool);
            }
            else
            {
                _conn_manager->deleteConnection(conn);
            }
        }
        // 需要完成的业务

        // 打开信道
        void OnOpenChannel(const muduo::net::TcpConnectionPtr &conn,
                           const openChannelRequestPtr &message,
                           muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 通过连接对象创建信道
            myconn->openChannel(message);
        }
        // 关闭信道
        void OncloseChannel(const muduo::net::TcpConnectionPtr &conn,
                            const closeChannelRequestPtr &message,
                            muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 通过连接对象关闭信道
            myconn->closeChannel(message);
        }
        // 声明交换机
        void OndeclareExchange(const muduo::net::TcpConnectionPtr &conn,
                               const declareExchangeRequestPtr &message,
                               muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("声明交换机时，信道不存在...");
                return;
            }
            channel->declareExchange(message);
        }
        // 移除交换机
        void OndeleteExchange(const muduo::net::TcpConnectionPtr &conn,
                              const deleteExchangeRequestPtr &message,
                              muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("移除交换机时，信道不存在...");
                return;
            }
            channel->deleteExchange(message);
        }
        // 声明队列
        void OndeclareQueue(const muduo::net::TcpConnectionPtr &conn,
                            const declareQueueRequestPtr &message,
                            muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("声明队列时，信道不存在...");
                return;
            }
            channel->declareQueue(message);
        }
        // 移除队列
        void OndeleteQueue(const muduo::net::TcpConnectionPtr &conn,
                           const deleteQueueRequestPtr &message,
                           muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("移除队列时，信道不存在...");
                return;
            }
            channel->deleteQueue(message);
        }
        // 队列的绑定
        void OnBindQueue(const muduo::net::TcpConnectionPtr &conn,
                         const queueBindRequestPtr &message,
                         muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("队列绑定时，信道不存在...");
                return;
            }
            channel->QueueBind(message);
        }
        // 队列解绑
        void OnUnBindQueue(const muduo::net::TcpConnectionPtr &conn,
                           const queueUnBindRequestPtr &message,
                           muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("队列解除绑定时，信道不存在...");
                return;
            }
            channel->QueueUnBind(message);
        }
        // 发布消息
        void OnBasicPublish(const muduo::net::TcpConnectionPtr &conn,
                            const basicPublishRequestPtr &message,
                            muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("发布消息时，信道不存在...");
                return;
            }
            channel->basicPublish(message);
        }
        // 对消息进行确认
        void OnBasicAck(const muduo::net::TcpConnectionPtr &conn,
                        const basicAckRequestPtr &message,
                        muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("对消息进行确认，信道不存在...");
                return;
            }
            // DLOG("broker模块,msg_id为:%s",message->msg_id().c_str());
            channel->basicAck(message);
        }
        // 订阅队列
        void OnBasicConsume(const muduo::net::TcpConnectionPtr &conn,
                            const basicConsumeRequestPtr &message,
                            muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("订阅队列时，信道不存在...");
                return;
            }
            channel->basicConsume(message);
        }
        // 取消订阅
        void OnBasicCancel(const muduo::net::TcpConnectionPtr &conn,
                           const basicCancelRequestPtr &message,
                           muduo::Timestamp)
        {
            // 先根据conn获取到对应的连接
            Connection::ptr myconn = _conn_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("收到非法的连接,已断开该连接\n");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr channel = myconn->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("取消订阅队列时，信道不存在...");
                return;
            }
            channel->basicCancel(message);
        }

        // 给分发器设置一个默认的消息处理，当请求消息的种类不存在时调用
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                              const MessagePtr &message,
                              muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

    private:
        muduo::net::EventLoop _baseloop;  // 事件循环
        muduo::net::TcpServer _TcpServer; // 服务器
        ProtobufDispatcher _dispatcher;   // 请求分发器对象，每一种请求对应一个回调方法
        ProtobufCodecPtr _codec;          // protobuf协议处理器，将收到的消息通过协议 进行序列化和反序列化
                                          // 再通过分发器完成业务层面的处理

        VirtualHost::ptr _host;
        ConnectionManager::ptr _conn_manager;
        ConsumerManager::ptr _consume_manager;
        ThreadPool::ptr _pool;
    };
}
#endif